package com.headfirst.dmp.tools

import com.headfirst.dmp.utils.{NBF, SchemaUtils}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.sql.types._

/**
  * 将Bzip文件转化成parquet文件
  *
  * 注意将数据保存成parquet文件是sparksql提供的功能
  */
object Bzip2Parquet {


  def main(args: Array[String]): Unit = {

    //1.判断参数的个数
    if (args.length != 3) {
      print(
        """
          |com.hedfirst.dmp.tools.Bzip2Parquet
          |参数：
          |    logInputPath
          |    ResultOutputPath
          |    compressionCodec  <snappy, gzip, lzo>
        """.stripMargin)
      sys.exit(-1) //非正常退出
    }

    //2.创建session对象
    val session: SparkSession = SparkSession.builder()
      .appName(s"${this.getClass.getSimpleName}")
      .master("local[*]")
      //设置压缩编码格式格式(snappy, gzip, lzo, 在Spark2.x后已经都默认snappy了，1.6等版本还是gzip)
      .config("spark.sql.parquet.compression.codec", args(2))
      //设置序列化方式采用KryoSerializer方式（默认的是java序列化）
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .getOrCreate()


    //3.读取文件，这里使用SparkContext对象去读取数据，为了后面给数据加上schema信息
    val file: RDD[String] = session.sparkContext.textFile(args(0))

    /**
      * 4.进行ETL处理
      */
    //过滤掉字段不够的
    //(参数-1表示一直切分到结尾，如果不加，则，，，这样的情况默认会忽略，导致少字段)
    val arrRDD: RDD[Array[String]] = file.map(_.split(",", -1)).filter(_.length >= 85)

    //创建schema信息
    val schema: StructType = SchemaUtils.structType

    //创建RDD[Row]
    val rdd: RDD[Row] = arrRDD.map(arr => {
      Row(
        arr(0),
        NBF.toInt(arr(1)),
        NBF.toInt(arr(2)),
        NBF.toInt(arr(3)),
        NBF.toInt(arr(4)),
        arr(5),
        arr(6),
        NBF.toInt(arr(7)),
        NBF.toInt(arr(8)),
        NBF.toDouble(arr(9)),
        NBF.toDouble(arr(10)),
        arr(11),
        arr(12),
        arr(13),
        arr(14),
        arr(15),
        arr(16),
        NBF.toInt(arr(17)),
        arr(18),
        arr(19),
        NBF.toInt(arr(20)),
        NBF.toInt(arr(21)),
        arr(22),
        arr(23),
        arr(24),
        arr(25),
        NBF.toInt(arr(26)),
        arr(27),
        NBF.toInt(arr(28)),
        arr(29),
        NBF.toInt(arr(30)),
        NBF.toInt(arr(31)),
        NBF.toInt(arr(32)),
        arr(33),
        NBF.toInt(arr(34)),
        NBF.toInt(arr(35)),
        NBF.toInt(arr(36)),
        arr(37),
        NBF.toInt(arr(38)),
        NBF.toInt(arr(39)),
        NBF.toDouble(arr(40)),
        NBF.toDouble(arr(41)),
        NBF.toInt(arr(42)),
        arr(43),
        NBF.toDouble(arr(44)),
        NBF.toDouble(arr(45)),
        arr(46),
        arr(47),
        arr(48),
        arr(49),
        arr(50),
        arr(51),
        arr(52),
        arr(53),
        arr(54),
        arr(55),
        arr(56),
        NBF.toInt(arr(57)),
        NBF.toDouble(arr(58)),
        NBF.toInt(arr(59)),
        NBF.toInt(arr(60)),
        arr(61),
        arr(62),
        arr(63),
        arr(64),
        arr(65),
        arr(66),
        arr(67),
        arr(68),
        arr(69),
        arr(70),
        arr(71),
        arr(72),
        NBF.toInt(arr(73)),
        NBF.toDouble(arr(74)),
        NBF.toDouble(arr(75)),
        NBF.toDouble(arr(76)),
        NBF.toDouble(arr(77)),
        NBF.toDouble(arr(78)),
        arr(79),
        arr(80),
        arr(81),
        arr(82),
        arr(83),
        NBF.toInt(arr(84))
      )
    })

    //5.将数据以parquet格式保存
    val frame: DataFrame = session.createDataFrame(rdd, schema)
    frame.write.parquet(args(1))

    //6.关闭流
    session.stop()

  }

}
